package com.ge.rpc.registry;

import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.ge.rpc.config.RegistryConfig;
import com.ge.rpc.model.ServiceMetaInfo;
import io.lettuce.core.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.vertx.core.impl.ConcurrentHashSet;

import java.time.Duration;
import java.util.*;

/**
 * Redis 服务注册中心
 */
public class RedisRegistry implements Registry {

    private RedisClient redisClient;

    private StatefulRedisConnection<String, String> statefulRedisConnection;

    private static final String REDIS_ROOT_PATH = "rpc:";

    /**
     * 本机注册的节点 key 集合（用于维护续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();

    /**
     * 注册中心服务缓存
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();

    /**
     * 正在监听的 Key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();

    @Override
    public void init(RegistryConfig registryConfig) {
        redisClient = RedisClient.create(RedisURI.builder()
                .withHost(registryConfig.getHost())
                .withPort(registryConfig.getPort())
                .withTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .withPassword(registryConfig.getPassword())
                .build());
        statefulRedisConnection = redisClient.connect();
        heartBeat();
    }

    @Override
    public void registry(ServiceMetaInfo serviceMetaInfo) {
        RedisCommands<String, String> redisCommands = statefulRedisConnection.sync();

        String registerKey = REDIS_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        // 存储节点信息，30秒过期
        redisCommands.setex(registerKey, 30, JSONUtil.toJsonStr(serviceMetaInfo));
        localRegisterNodeKeySet.add(registerKey);
    }

    @Override
    public void unRegistry(ServiceMetaInfo serviceMetaInfo) {
        RedisCommands<String, String> redisCommands = statefulRedisConnection.sync();
        String registerKey = REDIS_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        redisCommands.del(registerKey);
        localRegisterNodeKeySet.remove(registerKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 优先从缓存中获取服务
        List<ServiceMetaInfo> serviceMetaInfos = registryServiceCache.readCache();
        if (serviceMetaInfos != null) {
            return serviceMetaInfos;
        }

        List<ServiceMetaInfo> serviceMetaInfoList = new ArrayList<>();
        RedisCommands<String, String> redisCommands = statefulRedisConnection.sync();
        // 前缀查询
        ScanCursor scanCursor = ScanCursor.INITIAL;
        ScanArgs scanArgs = new ScanArgs().match(serviceKey + ":*");
        do {
            KeyScanCursor<String> scanned = redisCommands.scan(scanCursor, scanArgs);
            if (scanned != null) {
                List<String> keyList = scanned.getKeys();
                if (!keyList.isEmpty()) {
                    keyList.forEach(key -> {
                        // 增加 key 状态监听
                        watch(key);
                        serviceMetaInfoList.add(JSONUtil.toBean(redisCommands.get(key), ServiceMetaInfo.class));
                    });
                }
                scanCursor = scanned;
            } else {
                scanCursor = ScanCursor.FINISHED;
            }
        } while (!scanCursor.isFinished());
        // 写入服务缓存
        registryServiceCache.writeCache(serviceMetaInfoList);
        return serviceMetaInfoList;
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");
        RedisCommands<String, String> redisCommands = statefulRedisConnection.sync();
        for (String key : localRegisterNodeKeySet) {
            redisCommands.del(key);
        }
        statefulRedisConnection.close();
        redisClient.shutdown();
    }

    @Override
    public void heartBeat() {
        // 使用 Hutool 工具类的 CronUtil 实现定时任务，每 10 秒对所有集合中的节点执行一次重新注册
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                for (String key : localRegisterNodeKeySet) {
                    List<ServiceMetaInfo> serviceMetaInfos = serviceDiscovery(key);
                    if (serviceMetaInfos.isEmpty()) {
                        // 节点已过期，需要重启节点进行注册
                        continue;
                    }
                    // 节点未过期，重新注册
                    ServiceMetaInfo serviceMetaInfo = serviceMetaInfos.get(0);
                    registry(serviceMetaInfo);
                }
            }
        });
    }

    @Override
    public void watch(String serviceNodeKey) {
        //  发布/订阅连接
        StatefulRedisPubSubConnection<String, String> pubSubConnection = redisClient.connectPubSub();
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        // 防止重复监听同一个 key
        if (newWatch) {
            pubSubConnection.addListener(new RedisPubSubAdapter<String, String>() {
                @Override
                public void message(String channel, String message) {
                    if (message.equals("del")) {
                        registryServiceCache.clearCache();
                    }
                }
            });
            // 订阅特定 key 的事件
            pubSubConnection.sync().psubscribe("__keyspace@*__:" + serviceNodeKey);
            pubSubConnection.close();
        }

    }
}
